/**
 * 
 */
package edu.umd.clip.lm.storage;

import java.util.*;
import java.util.concurrent.locks.*;

/**
 * @author Denis Filimonov <den@cs.umd.edu>
 *
 */
public class AsyncConsumerMultiplexor<K, V> implements AsyncConsumer<K, V>, AsyncProducer<K, V> {
	private Lock lock;
	private HashMap<K,List<AsyncConsumer<K,V>>> consumers;
	private AsyncProducer<K,V> backend;
	
	
	AsyncConsumerMultiplexor(AsyncProducer<K, V> backend) {
		this.backend = backend;
		lock = new ReentrantLock();
		consumers = new HashMap<K,List<AsyncConsumer<K,V>>>();
	}

	/* (non-Javadoc)
	 * @see edu.umd.clip.lm.storage.AsyncConsumer#receivedData(java.lang.Object, java.lang.Object)
	 */
	@Override
	public void receivedData(K key, V value) {
		lock.lock();
		List<AsyncConsumer<K,V>> interestedConsumers = consumers.remove(key);
		lock.unlock();
		if (interestedConsumers != null) {
			for(AsyncConsumer<K,V> consumer : interestedConsumers) {
				consumer.receivedData(key, value);
			}
		}
	}

	/* (non-Javadoc)
	 * @see edu.umd.clip.lm.storage.AsyncProducer#request(java.lang.Object, edu.umd.clip.lm.storage.AsyncConsumer)
	 */
	@Override
	public void request(K key, AsyncConsumer<K, V> callback) {
		lock.lock();
		List<AsyncConsumer<K,V>> interestedConsumers = consumers.get(key);
		boolean isNew = false;
		if (interestedConsumers == null) {
			isNew = true;
			interestedConsumers = new ArrayList<AsyncConsumer<K,V>>(3);
			consumers.put(key, interestedConsumers);
		}
		interestedConsumers.add(callback);
		lock.unlock();
		
		if (isNew) {
			backend.request(key, this);
		}
	}
}
